package com.example.pulsar.service;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

/**
 * @Auther: 141040
 * @Date: 3/25/2019 10:09
 * @Description: Pulsar消费者
 */
@Service
public class ConsumersService {

    /**日志打印**/
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${pulsarUrl}")
    private String pulsarUrl;

    @Value("${pulsarTopic:order-topic}")
    private String pulsarTopic;

    /**
     * 创建消费者
     */
    public Consumer createConsumers() throws PulsarClientException {

        //1. 实例化PulsarClient客户端对象
        PulsarClient client = PulsarClient.builder().serviceUrl(pulsarUrl).build();

        //2. 实例化PulsarClient客户端对象
        String npTopic = "persistent://public/default/zhang-topic";
        String subscriptionName = "order-subscription";
        Consumer consumer = client.newConsumer()
                .topic(npTopic)
                .subscriptionName(subscriptionName)
                .ackTimeout(3600, TimeUnit.SECONDS)
                .subscribe();
        return consumer;

    }





}
